package kafka;

import com.alibaba.fastjson.JSON;
import kafka.Pojo.FileBeat;
import kafka.Pojo.NginxLog;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerMain {
    public static void main(String[] args) {
        //设置kafka消费者配置
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.50.188.227:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"nginx");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //创建kafka消费者实例
        KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

        //订阅主题
        consumer.subscribe(Collections.singletonList("nginx-log"));



        //消费消息
        try{
            while (true){
                //拉取消息
                ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));
                for(ConsumerRecord<String,String> record: records){
                    //System.out.println(record.value());
                    String log = record.value();
                    FileBeat fileBeat = JSON.parseObject(log, FileBeat.class);
                    NginxLog nginxLog = JSON.parseObject(fileBeat.getMessage(), NginxLog.class);
                    System.out.println(nginxLog.toString());
                }
            }
        } finally {
            consumer.close();
        }

    }
}
